package org.niit.stream

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.util.Random

/*
 自定义Receiver的Streaming
  Spark Streaming 可以实时的接收 套接字 队列 HDFS HBase等数据流，唯独像MySQL不能实时接收数据，面对这种情况需要自定义Receiver
 */
object SparkStreaming_03 {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("streaming")
    //sparkConf.set("spark.testing.memory","4442528585")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //创建 自定义的receiver Streaming
    val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver)
    messageDS.print()

    ssc.start()
    ssc.awaitTermination()
  }
  //自定义数据采集器
  //重写方法 onStart onStop
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {

    private var flag = true

    //启动采集的方法
    override def onStart(): Unit = {

      new Thread(new Runnable {
        override def run(): Unit = {
          while (flag){

            val message = "采集的数据为："+ new Random().nextInt(10).toString
            //返回采集到的数据  ，这里不能用return进行返回。 因为return带有结束方法的作用
            store(message)
            Thread.sleep(2000)
          }
        }
      }).start()

    }
    //关闭采集的方法
    override def onStop(): Unit = {
      flag = false
    }
  }



}
